package com.xiehua.task.job.bucket;

import com.xiehua.task.actor.cmd.NotifyClient;
import com.xiehua.task.job.RidesOperationDefualt;
import org.tio.core.Aio;
import org.tio.core.ChannelContext;
import org.tio.core.GroupContext;
import org.tio.core.Node;
import redis.clients.jedis.Jedis;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

public class TopicBucket extends RidesOperationDefualt {

    public static final String TOPIC_INFO_PIX = "topic_info_";//--->hash_topname(ip+port,cid)

    public static final String TOPIC_INFO_POOL = "topic_pool";//--->set<topname>

    public TopicBucket(Jedis jedis) {
        super(jedis);
    }


    /**
     * 维护topic和客户端信息
     **/
    public void addTopicInfo(String topicName, Node node, String cid,GroupContext context) {
        //1.check and add top_name
        if (!jedis.sismember(TOPIC_INFO_POOL, topicName)) jedis.sadd(TOPIC_INFO_POOL, topicName);
        //2.remove invalid client
        jedis.hgetAll(TOPIC_INFO_PIX + topicName).entrySet().forEach(s ->{
            if(s.getKey().startsWith("$")) jedis.hdel(TOPIC_INFO_PIX + topicName,s.getKey());
            String[] strings = s.getKey().split(":");
            if (strings == null || strings.length < 1) return ;
            ChannelContext channelContext = Aio.getChannelContextByClientNode(context, strings[0], Integer.parseInt(strings[1]));
            if (channelContext == null || channelContext.isWaitingClose() || channelContext.isRemoved() || channelContext.isClosed()) jedis.hdel(TOPIC_INFO_PIX + topicName,s.getKey());
        });
        //3.add client info
        String filed = node.getIp() + ":" + node.getPort();
        if (!jedis.hexists(TOPIC_INFO_PIX + topicName, filed)) jedis.hset(TOPIC_INFO_PIX + topicName, filed, cid);
    }

    /**
     * 查询所有topic名字
     **/
    public Set<String> queryTopics() {
        return jedis.smembers(TOPIC_INFO_POOL);
    }

    /**
     * 客户端下线时,移除对应的top_info(ip,port)
     **/
    public void delTopicInfo(Node node, GroupContext context) {
        String filed = node.getIp() + ":" + node.getPort();
        queryTopics().stream().map(s -> {
            if (jedis.hexists(TOPIC_INFO_PIX + s, filed)) {
                jedis.hdel(TOPIC_INFO_PIX + s, filed);
                return s;
            }
            return null;
        }).filter(m -> m != null).forEach(t ->{
            List<NotifyClient> clients = querySurvivalChannelByTopicName(t, context);
            if (clients == null || clients.size() < 1) {//没有存活的节点
                delTopicFromPool(t);
            }
        });
    }

    /**
     * 从topic pool移除一条数据
     **/

    public void delTopicFromPool(String topicName) {
        jedis.srem(TOPIC_INFO_POOL, topicName);
    }

    /**
     * 根据node查询关联的topic
     **/
    public List<String> queryTopic(Node node) {
        final String filed = node.getIp() + ":" + node.getPort();
        return queryTopics().stream().map(s -> {
            if (jedis.hexists(TOPIC_INFO_PIX + s, filed)) return s;
            return null;
        }).filter(t -> t != null).collect(Collectors.toList());
    }

    /**
     * 根据topName查询存活的节点
     **/
    public List<NotifyClient> querySurvivalChannelByTopicName(String topicName, GroupContext context) {
        Map<String, String> map = jedis.hgetAll(TOPIC_INFO_PIX + topicName);
        if (map == null || map.keySet().size() < 1) return null;
        //query survival channel by ip and port
        return map.entrySet().stream().map(s -> {
            String key = s.getKey();
            String[] strings = key.split(":");
            if (strings == null || strings.length < 1) return null;
            ChannelContext channelContext = Aio.getChannelContextByClientNode(context, strings[0], Integer.parseInt(strings[1]));
            if (channelContext == null || channelContext.isWaitingClose()) return null;
            return new NotifyClient(topicName, strings[0], Integer.parseInt(strings[1]), s.getValue());

        }).filter(t -> t != null).collect(Collectors.toList());
    }


}
